/*
 * Copyright © 2020-present zmzhou-star. All Rights Reserved.
 */

package com.handa.admin.userservice.websocket.handle;

import com.alibaba.fastjson.JSONObject;
import com.handa.admin.userservice.entity.VO.OnlineSysUserVO;
import com.handa.admin.userservice.websocket.entity.WebsocketDataBody;
import com.handa.admin.webChat.entity.ChatMessageDO;
import com.handa.admin.webChat.service.ChatMessageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.handa.admin.userservice.websocket.Enums.WebSocketDataTypeEnum.*;


/**
 * WebSocket处理器
 *
 * @author zmzhou
 * @version 1.0
 * @title WebShellWebSocketHandler
 * @date 2021/2/22 20:58
 */
@Slf4j
@Component
public class GlobalWebSocketHandler extends TextWebSocketHandler {

    // 某个用户名 对应多个在线记录
    public static final ConcurrentHashMap<String, Set<WebSocketSession>> map_userName_sessionSet = new ConcurrentHashMap<>();

    // sessionId: Session
    public static final ConcurrentHashMap<String, WebSocketSession> map_id_session = new ConcurrentHashMap<>();


    public static final ConcurrentHashMap<String, OnlineSysUserVO> map_session_onlineUser = new ConcurrentHashMap<>();


    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);


    // 创建SimpleDateFormat对象，指定日期格式
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Resource
    private ChatMessageService chatMessageService;

    @PostConstruct
    public void init() {
        // 每分钟发送系统时间, 也作为心跳，检测前端是否在线。
        scheduler.scheduleAtFixedRate(this::broadcastSystemTime, 0, 1, TimeUnit.MINUTES);
    }

    @Override
    protected void handleTextMessage(@NotNull WebSocketSession session, TextMessage message) throws IllegalAccessException {
        String payload = message.getPayload();
        log.info("receive message :{}", payload);
        WebsocketDataBody websocketDataBody = JSONObject.parseObject(payload, WebsocketDataBody.class);

        Integer _msgType = websocketDataBody.getType();
        if (_msgType == null) {
            log.error("发送的数据体有误，不包含消息类型数据");
            return;
        }
        // webChat 用户发来的消息
        if (_msgType == WEB_CHAT_USER_TEXT_MESSAGE.code) {
            handleReceiveWebChatOneUserTextMessage(session, websocketDataBody);
        } else if (_msgType == WEB_CHAT_USER_EMOJI_MESSAGE.code) {
            handleReceiveWebChatOneUserEMOJMessage(session, websocketDataBody);
        } else if (_msgType == WEB_CHAT_USER_IMAGE_MESSAGE.code) {
            handleReceiveWebChatOneUserImageMessage(session, websocketDataBody);
        } else if (_msgType == WEB_CHAT_USER_FILE_MESSAGE.code) {
            handleReceiveWebChatOneUserFileMessage(session, websocketDataBody);
        } else if (_msgType == SYSTEM_USER_ONLINE_MESSAGE.code) {
            handleUserOnlineMessage(session, websocketDataBody);
        } else {
            log.error("未处理的消息类型");
        }

    }

    // 转发 用户上线信息 给所有Session
    private void handleUserOnlineMessage(WebSocketSession session, WebsocketDataBody websocketDataBody) {
        websocketDataBody.setFromSessionId(session.getId());
        OnlineSysUserVO user = new OnlineSysUserVO();
        user.setUserId(websocketDataBody.getFromUserId());
        user.setNickname(websocketDataBody.getFromNickName());
        user.setSessionId(session.getId());
        user.setUsername(websocketDataBody.getFromUserName());

        map_session_onlineUser.put(session.getId(), user);

        broadcastMessage(websocketDataBody);
        logOnlineUserInfo();
    }

    // 转发某个用户通过 webChat 发送的消息
    private void handleReceiveWebChatOneUserTextMessage(WebSocketSession session, WebsocketDataBody websocketDataBody) {
        // 如果没有传 接收人就默认广播
        String toUserId = websocketDataBody.getToUserId();
        if (StringUtils.isBlank(toUserId)) {
            websocketDataBody.setToUserId("-1");
        }


        ChatMessageDO chatMessageDO = new ChatMessageDO();
        chatMessageDO.setMessageType(0);
        chatMessageDO.setContent(JSONObject.toJSONString(websocketDataBody));
        chatMessageDO.setCreatedAt(websocketDataBody.getCreateTime());
        chatMessageDO.setSender(websocketDataBody.getFromUserId());
        chatMessageDO.setReceiver(websocketDataBody.getToUserId());

        chatMessageService.save(chatMessageDO);

        forwardWebChatMessageToOthers(websocketDataBody, session.getId());
    }

    // 转发某个用户通过 webChat 发送的 表情 消息
    private void handleReceiveWebChatOneUserEMOJMessage(WebSocketSession session, WebsocketDataBody websocketDataBody) {

        // 如果没有传 接收人就默认广播
        String toUserId = websocketDataBody.getToUserId();
        if (StringUtils.isBlank(toUserId)) {
            websocketDataBody.setToUserId("-1");
        }


        ChatMessageDO chatMessageDO = new ChatMessageDO();
        chatMessageDO.setMessageType(WEB_CHAT_USER_TEXT_MESSAGE.code);
        chatMessageDO.setContent(JSONObject.toJSONString(websocketDataBody));
        chatMessageDO.setCreatedAt(websocketDataBody.getCreateTime());
        chatMessageDO.setSender(websocketDataBody.getFromUserId());
        chatMessageDO.setReceiver(websocketDataBody.getToUserId());

        chatMessageService.save(chatMessageDO);

        forwardWebChatMessageToOthers(websocketDataBody, session.getId());
    }

    // 转发某个用户通过 webChat 发送的 图片 消息
    private void handleReceiveWebChatOneUserImageMessage(WebSocketSession session, WebsocketDataBody websocketDataBody) {
        // 如果没有传 接收人就默认广播
        String toUserId = websocketDataBody.getToUserId();
        if (StringUtils.isBlank(toUserId)) {
            websocketDataBody.setToUserId("-1");
        }


        ChatMessageDO chatMessageDO = new ChatMessageDO();
        chatMessageDO.setMessageType(WEB_CHAT_USER_IMAGE_MESSAGE.code);
        chatMessageDO.setContent(JSONObject.toJSONString(websocketDataBody));
        chatMessageDO.setCreatedAt(websocketDataBody.getCreateTime());
        chatMessageDO.setSender(websocketDataBody.getFromUserId());
        chatMessageDO.setReceiver(websocketDataBody.getToUserId());

        chatMessageService.save(chatMessageDO);

        forwardWebChatMessageToOthers(websocketDataBody, session.getId());
    }

    // 转发某个用户通过 webChat 发送的 文件 消息
    private void handleReceiveWebChatOneUserFileMessage(WebSocketSession session, WebsocketDataBody websocketDataBody) throws IllegalAccessException {
        // 如果没有传 接收人就默认广播
        String toUserId = websocketDataBody.getToUserId();
        if (StringUtils.isBlank(toUserId)) {
            websocketDataBody.setToUserId("-1");
        }


        ChatMessageDO chatMessageDO = new ChatMessageDO();
        chatMessageDO.setMessageType(WEB_CHAT_USER_FILE_MESSAGE.code); // 1 为文件类型
        chatMessageDO.setContent(JSONObject.toJSONString(websocketDataBody));
        chatMessageDO.setCreatedAt(websocketDataBody.getCreateTime());
        chatMessageDO.setSender(websocketDataBody.getFromUserId());
        chatMessageDO.setReceiver(websocketDataBody.getToUserId());

        chatMessageService.save(chatMessageDO);

        forwardWebChatMessageToOthers(websocketDataBody, session.getId());
    }
    //

    /**
     * 发送消息给别的Session
     */
    private void forwardWebChatMessageToOthers(WebsocketDataBody websocketDataBody, String excludeSessionId) {

        map_id_session.forEach((sessionId, session) -> {
            try {

                if (StringUtils.isNotBlank(excludeSessionId)) {
                    if (sessionId.equals(excludeSessionId)) {
                        return;
                    }
                }


                String sendMsgStr = JSONObject.toJSONString(websocketDataBody);
                log.info("send message :{}", sendMsgStr);

                session.sendMessage(new TextMessage(sendMsgStr));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }

    private void broadcastMessage(WebsocketDataBody websocketDataBody) {

        map_id_session.forEach((sessionId, session) -> {
            try {
                if (!session.isOpen()) {
                    return;
                }
                String sendMsgStr = JSONObject.toJSONString(websocketDataBody);
                log.info("send message :{}", sendMsgStr);

                session.sendMessage(new TextMessage(sendMsgStr));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }

    public void logOnlineUserInfo() {
        log.info("------------------------------在线信息----------------------------");
        log.info("map_id_session: " + JSONObject.toJSONString(map_id_session.size()));
        log.info("map_session_onlineUser: " + JSONObject.toJSONString(map_session_onlineUser.values()));
        log.info("map_userName_sessionSet: " + JSONObject.toJSONString(map_userName_sessionSet.size()));
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        Map<String, Object> attributes = session.getAttributes();
        Object _username = attributes.get("username");
        String username = "";
        if (!Objects.isNull(_username)) {
            username = (String) _username;
            Set<WebSocketSession> sessions = map_userName_sessionSet.get(username);
            if (sessions == null) {

                Set<WebSocketSession> userSession = new HashSet<>();
                userSession.add(session);
                map_userName_sessionSet.put(username, userSession);
                map_id_session.put(session.getId(), session);

            } else {
                sessions.add(session);
            }

        }

        log.info("webSocket 新连接: " + session.getId() + attributes.get("ipAddress") + "  " + username);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String sessionId = session.getId();
        // 删除 map_userName_session
        removeSessionById(sessionId, map_userName_sessionSet);

        // 删除 map_id_session
        map_id_session.remove(sessionId);

        OnlineSysUserVO onlineUser = map_session_onlineUser.get(sessionId);
        if (onlineUser != null) {
            WebsocketDataBody websocketDataBody = new WebsocketDataBody();
            websocketDataBody.setType(SYSTEM_USER_OFFLINE_MESSAGE.code);
            websocketDataBody.setFromUserId(onlineUser.getUserId());
            websocketDataBody.setFromNickName(onlineUser.getNickname());
            websocketDataBody.setFromUserName(onlineUser.getUsername());
            broadcastMessage(websocketDataBody);
        }


        map_session_onlineUser.remove(sessionId);

        log.info("webSocket 断开: " + sessionId + ", Status: " + status);
        logOnlineUserInfo();
    }

    private void removeSessionById(String sessionId, Map<String, Set<WebSocketSession>> sessionMap) {
        Iterator<Map.Entry<String, Set<WebSocketSession>>> mapIterator = sessionMap.entrySet().iterator();

        while (mapIterator.hasNext()) {
            Map.Entry<String, Set<WebSocketSession>> entry = mapIterator.next();
            Set<WebSocketSession> sessions = entry.getValue();

            Iterator<WebSocketSession> sessionIterator = sessions.iterator();
            while (sessionIterator.hasNext()) {
                if (sessionIterator.next().getId().equals(sessionId)) {
                    sessionIterator.remove();
                    if (sessions.isEmpty()) {
                        mapIterator.remove();
                    }
                    return; // 找到并移除后立即返回
                }
            }
        }
    }

    // 广播发送时间，给前端用于前端校验系统时间
    private void broadcastSystemTime() {
        synchronized (map_id_session) {

            WebsocketDataBody websocketDataBody = new WebsocketDataBody();
            websocketDataBody.setType(SYSTEM_BROADCAST_TIME_MESSAGE.code);

            Map<String, Object> timeMessage = new HashMap<>();
            timeMessage.put("serverTime", System.currentTimeMillis());
            timeMessage.put("formatTime", sdf.format(new Date()));

            websocketDataBody.setData(timeMessage);

            broadcastMessage(websocketDataBody);

        }
    }

}
